package com.example.demo.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Optional;

import static com.example.demo.config.RabbitConfig.DELAYED_QUEUE_NAME;

@Slf4j
@Component
public class MessageConsumer {

    // 最大重试次数常量
    private static final int MAX_RETRY_COUNT = 3;

    @RabbitListener(queues = DELAYED_QUEUE_NAME, ackMode = "MANUAL")
    public void receiveMessage(String message, Message amqpMessage, Channel channel) throws IOException {
        long deliveryTag = amqpMessage.getMessageProperties().getDeliveryTag();
        try {
            log.info("收到消息: {}", message);
            
            // 使用Optional处理消息
            // 旧代码：processMessage(message);
            Optional.ofNullable(message)
                .ifPresent(this::processMessage);
            
            // 手动确认消息
            channel.basicAck(deliveryTag, false);
            log.info("消息处理成功，已确认: {}", message);
            
        } catch (Exception e) {
            log.error("消息处理失败: {}", e.getMessage());
            
            // 使用Optional获取重试次数
            // 旧代码：
            // Integer retryCount = amqpMessage.getMessageProperties().getHeader("retry-count");
            // if (retryCount == null) {
            //     retryCount = 0;
            // }
            int retryCount = Optional.ofNullable(amqpMessage.getMessageProperties().getHeader("retry-count"))
                .map(count -> (Integer) count)
                .orElse(0);
            
            if (retryCount < MAX_RETRY_COUNT) {
                // 拒绝消息并重新入队
                channel.basicNack(deliveryTag, false, true);
                log.info("消息已重新入队，准备重试，当前重试次数: {}", retryCount + 1);
            } else {
                // 拒绝消息并丢弃
                channel.basicReject(deliveryTag, false);
                log.error("消息重试次数超过最大限制，已被丢弃");
            }
        }
    }
    
    private void processMessage(String message) {
        // 使用Optional处理消息内容检查
        // 旧代码：
        // if (message.contains("error")) {
        //     throw new RuntimeException("消息处理失败");
        // }
        Optional.ofNullable(message)
            .filter(msg -> !msg.contains("error"))
            .orElseThrow(() -> new RuntimeException("消息处理失败"));

        // 模拟处理耗时
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}